/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wx.mq.startup;

import cn.ac.iie.di.commons.httpserver.framework.server.HttpServer;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import java.util.HashMap;
import java.util.Timer;
import org.apache.log4j.Logger;
import cn.ac.iie.di.wx.mq.common.RuntimeEnv;
import static cn.ac.iie.di.wx.mq.common.RuntimeEnv.retryQueue;
import cn.ac.iie.di.wx.mq.configuration.Configuration;
import cn.ac.iie.di.wx.mq.killHandler.KillHandler;
import cn.ac.iie.di.wx.mq.mqservice.Receiver;
import cn.ac.iie.di.wx.mq.redisThread.JedisThread;
import cn.ac.iie.di.wx.mq.timerTask.CountTask;
import cn.ac.iie.di.wx.mq.timerTask.RedoTask;
import cn.ac.iie.di.wx.mq.timerTask.RetryFileThread;
import cn.ac.iie.di.wx.mq.utils.RedisUtil;
import cn.ac.iie.di.wx.mq.utils.TimeStampUtil;
import cn.ac.iie.di.wx.mq.utils.ToJsonUtil;
import java.util.ArrayList;
import java.util.List;

/**
 *
 * @author Administrator
 */
public class MqStartup {

    static Logger logger = Logger.getLogger(MqStartup.class.getName());
    static HttpServer httpServer = null;

    public static void main(String[] args) {
        try {
            init();
            logger.info(RuntimeEnv.getParam(RuntimeEnv.IP));
            logger.info(RuntimeEnv.getParam(RuntimeEnv.PORT));
            //从消息队列中拉取群聊多媒体请求的字符串
            TimeStampUtil stampUtil = new TimeStampUtil();
            REConnection reconntction = mqThreadStartup(stampUtil);
            REConnection urgent_mqreconntction = urgent_mqThreadStartup(stampUtil);
            //将相关的字段更新到reids
            Thread.sleep(1000);
            redisThreadStartup();
            retryThreadStartup();//将失败的信息输入到文件中，由定时任务进行发送
            //统计
            CountTask countTask = new CountTask();
            Timer timer = new Timer();
            timer.schedule(countTask, 1000, 1000);
            //每隔一段时间进行重发
            Timer redoTimer = new Timer();
            RedoTask redoTask = new RedoTask();
            redoTimer.schedule(redoTask, 1000, 1000 * 60 * 60);

            KillHandler killHandler = new KillHandler();
            killHandler.registerSignal("TERM");
            List<REConnection> conns = new ArrayList<>();
            conns.add(reconntction);
            conns.add(urgent_mqreconntction);
            killHandler.setConns(conns);
            killHandler.putQueue("redisQueue", RuntimeEnv.redisQueue);
            killHandler.putQueue("urgent_redisQueue", RuntimeEnv.urgent_redisQueue);
            killHandler.putQueue("retryQueue", RuntimeEnv.retryQueue);
            logger.info("start mq at " + RuntimeEnv.getParam(RuntimeEnv.IP) + " success.");
        } catch (Exception ex) {
            logger.fatal(ex.getMessage(), ex);
            System.exit(-1);
        }

    }

    public static void init() throws Exception {
        String configurationFileName = "configuration.properties";
        Configuration conf = Configuration.getConfiguration(ClassLoader.getSystemClassLoader().getResource(configurationFileName).getPath());
        if (conf == null) {
            logger.error("reading " + configurationFileName + " is failed.");
        }
        try {
            RuntimeEnv.initialize(conf);///参数，
            logger.info("init successfully");
        } catch (Exception ex) {
            logger.error("initializng runtime enviroment is failed for" + ex.getMessage());
        }
    }

    public static void redisThreadStartup() {
        JedisThread jedisThread = new JedisThread(RuntimeEnv.redisQueue, RuntimeEnv.urgent_redisQueue);
        Thread thread = new Thread(jedisThread);
        thread.start();
    }

    public static void retryThreadStartup() {

        RetryFileThread retryTask = new RetryFileThread();
        Thread t = new Thread(retryTask);
        t.start();
    }

    public static REConnection mqThreadStartup(TimeStampUtil stampUtil) {
        try {
            Receiver receiver = new Receiver((String) RuntimeEnv.getParam(RuntimeEnv.mq_reconnection),
                    (String) RuntimeEnv.getParam(RuntimeEnv.mq_topic),
                    (String) RuntimeEnv.getParam(RuntimeEnv.mq_groupName),
                    Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.mq_threadNum)),
                    stampUtil,
                    RuntimeEnv.redisQueue);
            receiver.startSession();
            logger.info("start mq,host:" + (String) RuntimeEnv.getParam(RuntimeEnv.mq_reconnection)
                    + ",topic:" + (String) RuntimeEnv.getParam(RuntimeEnv.mq_topic)
                    + ",groupName:" + (String) RuntimeEnv.getParam(RuntimeEnv.mq_groupName)
                    + ",threadNum:" + (String) RuntimeEnv.getParam(RuntimeEnv.mq_threadNum));
            return receiver.getConn();
        } catch (REConnectionException ex) {
            logger.error(ex.getMessage(), ex);
            return null;
        }
    }

    public static REConnection urgent_mqThreadStartup(TimeStampUtil stampUtil) {
        try {
            Receiver receiver = new Receiver((String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_reconnection),
                    (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_topic),
                    (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_groupName),
                    Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_threadNum)),
                    stampUtil,
                    RuntimeEnv.urgent_redisQueue);
            receiver.startSession();
            logger.info("start urgent mq,host:" + (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_reconnection)
                    + ",topic:" + (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_topic)
                    + ",groupName:" + (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_groupName)
                    + ",threadNum:" + (String) RuntimeEnv.getParam(RuntimeEnv.urgent_mq_threadNum));
            return receiver.getConn();
        } catch (REConnectionException ex) {
            logger.error(ex.getMessage(), ex);
            return null;
        }
    }
}
